package com.iot.ig.mqtt.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;
import java.util.UUID;

@Configuration
public class MqttConfig {

    @Resource
    private MqttClient mqttClient;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();

        // mqtt服务地址，可以是多个
        options.setServerURIs(new String[]{mqttClient.getHost()});
        options.setUserName(mqttClient.getUsername());
        options.setPassword(mqttClient.getPassword().toCharArray());
        factory.setConnectionOptions(options);

        return factory;
    }

    /**
     * 2、接收消息的通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter() {
        // 订阅主题,保证唯一性
        String inClientId = UUID.randomUUID().toString().replace("-", "");
        String[] topic = {};

        return new MqttPahoMessageDrivenChannelAdapter(
                inClientId,
                mqttClientFactory(),
                topic);
    }

    /**
     * 接收消息
     *
     * @return
     */
    @Bean
    public MessageProducer inbound() {
        // 最后的#相当于通配符的概念
//        String[] topic = {"topic_prefix/topic/#"};
        // 默认空,在实际环境中，这个topic应该由客户端自己定义,这里只是给定了参数
        MqttPahoMessageDrivenChannelAdapter adapter = this.mqttPahoMessageDrivenChannelAdapter();

        adapter.setCompletionTimeout(5000);

        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        // 按字节接收消息
//        defaultPahoMessageConverter.setPayloadAsBytes(true);
        adapter.setConverter(defaultPahoMessageConverter);
        // 设置QoS
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());

        return adapter;
    }

    /**
     * 发送消息的通道
     *
     * @return
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 发送消息
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler outbound() {

        // 连接clientId保证唯一
        String outClientId = UUID.randomUUID().toString().replace("-", "");

        // 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(outClientId, mqttClientFactory());
        // 如果设置成true，即异步，发送消息时将不会阻塞。
        // messageHandler.setAsync(true);
        // 设置默认的topic
        // messageHandler.setDefaultTopic("defaultTopic");
        // 设置默认QoS
        messageHandler.setDefaultQos(0);

        // Paho消息转换器
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();

        // 发送默认按字节类型发送消息
        // defaultPahoMessageConverter.setPayloadAsBytes(true);
        messageHandler.setConverter(defaultPahoMessageConverter);
        return messageHandler;
    }
}
